/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.nageoffer.shortlink.project.config;

import com.nageoffer.shortlink.project.mq.consumer.ShortLinkStatsSaveConsumer;
import lombok.RequiredArgsConstructor;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.data.redis.stream.Subscription;

import java.time.Duration;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import static com.nageoffer.shortlink.project.common.constant.RedisKeyConstant.SHORT_LINK_STATS_STREAM_GROUP_KEY;
import static com.nageoffer.shortlink.project.common.constant.RedisKeyConstant.SHORT_LINK_STATS_STREAM_TOPIC_KEY;

/**
 * Redis Stream 消息队列配置
 */
@Configuration
@RequiredArgsConstructor
public class RedisStreamConfiguration {

    private final RedisConnectionFactory redisConnectionFactory;
    private final ShortLinkStatsSaveConsumer shortLinkStatsSaveConsumer;

    /**
     * 创建一个用于异步处理流数据消费的线程池
     * 该线程池主要用于处理短链接统计的异步流数据消费任务
     *
     * @return ExecutorService 一个线程池实例，用于执行异步任务
     */
    @Bean
    public ExecutorService asyncStreamConsumer() {
        // 创建一个原子整数，用于线程命名的唯一标识
        AtomicInteger index = new AtomicInteger();

        // 创建并返回一个线程池实例
        return new ThreadPoolExecutor(
            1, // 核心线程数为1，保持线程池的最小规模
            1, // 最大线程数也为1，限制线程池的最大规模
            60, // 空闲线程存活时间为60秒，即空闲线程在60秒后会被终止
            TimeUnit.SECONDS, // 时间单位为秒
            new SynchronousQueue<>(), // 使用同步队列，直接将任务提交给线程执行，不保持任务队列
            runnable -> {
                // 自定义线程工厂，为每个线程命名并设置为守护线程
                Thread thread = new Thread(runnable);
                thread.setName("stream_consumer_short-link_stats_" + index.incrementAndGet());
                thread.setDaemon(true);
                return thread;
            }, // 线程工厂，用于创建线程并设置线程属性
            new ThreadPoolExecutor.DiscardOldestPolicy() // 拒绝策略，丢弃最旧的任务并尝试重新提交当前任务
        );
    }

    /**
     * 创建并配置一个短链统计信息保存消费者的订阅
     *
     * @param asyncStreamConsumer 异步处理消息的执行器服务
     * @return Subscription对象，表示消费者对短链统计流的订阅
     */
    @Bean
    public Subscription shortLinkStatsSaveConsumerSubscription(ExecutorService asyncStreamConsumer) {
        // 配置流消息监听容器的选项
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options =
                StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                        .builder()
                        // 一次最多获取多少条消息
                        .batchSize(10)
                        // 执行从 Stream 拉取到消息的任务流程
                        .executor(asyncStreamConsumer)
                        // 如果没有拉取到消息，需要阻塞的时间。不能大于 ${spring.data.redis.timeout}，否则会超时
                        .pollTimeout(Duration.ofSeconds(3))
                        .build();

        // 构建读取流的请求
        StreamMessageListenerContainer.StreamReadRequest<String> streamReadRequest =
                StreamMessageListenerContainer.StreamReadRequest.builder(StreamOffset.create(SHORT_LINK_STATS_STREAM_TOPIC_KEY, ReadOffset.lastConsumed()))
                        // 在出现错误时是否取消订阅，这里选择不取消
                        .cancelOnError(throwable -> false)
                        // 指定消费者组和消费者名称
                        .consumer(Consumer.from(SHORT_LINK_STATS_STREAM_GROUP_KEY, "stats-consumer"))
                        // 自动确认消息处理成功
                        .autoAcknowledge(true)
                        .build();

        // 创建流消息监听容器
        StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = StreamMessageListenerContainer.create(redisConnectionFactory, options);

        // 在容器中注册流读取请求和消费者
        Subscription subscription = listenerContainer.register(streamReadRequest, shortLinkStatsSaveConsumer);

        // 启动监听容器开始消费消息
        listenerContainer.start();

        // 返回订阅对象，以便外部管理或使用
        return subscription;
    }
}
